package spring.cloud.tasks.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;

import java.util.concurrent.atomic.AtomicBoolean;

import static spring.cloud.tasks.common.Threads.sleep;

/**
 * @author huihui
 */
@Slf4j
@Setter
@Getter
public abstract class ConnectionReConnectListener implements ConnectionStateListener {
    private String executorId;
    private long sessionId;
    private volatile boolean connected = true;//连接zk成功后再添加ConnectionStateListener
    private volatile boolean closed = false;
    /**
     * <pre>
     * zk客户端sessionId发生了改变,会话上面的锁和临时节点信息被zk服务器端删除。所以必须重启执行器(相关的组件)。调用此方法通知外部程序重新启动执行器。
     * ConnectionState.SUSPENDED == connectionState 启动一个监听会话id变更线程，只要会话id改变。或者ConnectionState.LOST会话丢失都会进行执行器重启。
     * 这里使用cas进行一次重启操作。
     * </pre>
     */
    private volatile AtomicBoolean restartExecutored = new AtomicBoolean(false);

    public ConnectionReConnectListener(String executorId, long sessionId) {
        this.executorId = executorId;
        this.sessionId = sessionId;
    }

    public void close() {
        this.closed = true;
    }

    /**
     * 本类里面涉及的文档注释和普通注释都很重要,希望不要删除掉 add by huihui
     * <pre>
     * copy from https://www.jianshu.com/p/7149e1a6986f
     * void validateConnection(Watcher.Event.KeeperState state)
     * {
     *     if ( state == Watcher.Event.KeeperState.Disconnected )
     *     {
     *         suspendConnection();//任何一台服务器都没有连接上，属于闪断那种 后面可以还可以通过重新使用相同的连接连接上。
     *     }
     *     else if ( state == Watcher.Event.KeeperState.Expired )
     *     {
     *         connectionStateManager.addStateChange(ConnectionState.LOST);//会话过期,生成一个新的会话id
     *     }
     *     else if ( state == Watcher.Event.KeeperState.SyncConnected )
     *     {
     *         connectionStateManager.addStateChange(ConnectionState.RECONNECTED);
     *     }
     *     else if ( state == Watcher.Event.KeeperState.ConnectedReadOnly )
     *     {
     *         connectionStateManager.addStateChange(ConnectionState.READ_ONLY);
     *     }
     * }
     *
     * </pre>
     * <pre>
     * Watcher.Event.KeeperState
     * [需要注意的状态]SyncConnected:The client is in the connected state - it is connected to a server in the ensemble (one of the servers specified in the host connection parameter during ZooKeeper client creation).
     * [需要注意的状态:连接过期需要重新连接]Expired:The serving cluster has expired this session. The ZooKeeper client connection (the session) is no longer valid. You must create a new client connection (instantiate a new ZooKeeper instance) if you with to access the ensemble.
     * [需要注意的状态]Disconnected:The client is in the disconnected state - it is not connected to any server in the ensemble.
     * AuthFailed:Auth failed state
     * ConnectedReadOnly:The client is connected to a read-only server, that is the server which is not currently connected to the majority. The only operations allowed after receiving this state is read operations. This state is generated for read-only clients only since read/write clients aren't allowed to connect to r/o servers.
     * SaslAuthenticated:SaslAuthenticated: used to notify clients that they are SASL-authenticated, so that they can perform Zookeeper actions with their SASL-authorized permissions.
     * </pre>
     */
    @Override
    public void stateChanged(final CuratorFramework curatorFramework, final ConnectionState connectionState) {
        if (closed) {
            return;
        }
        if (ConnectionState.CONNECTED == connectionState) {
            /**
             * (copy from org.apache.curator.framework.state.ConnectionState)
             * Sent for the first successful connection to the server.
             * NOTE: You will only get one of these messages for any CuratorFramework instance.
             *
             * ConnectionState.CONNECTED=初始连接状态
             */
            log.debug("最开始连接zk成功后的状态,可忽略");
            //connected = true
        } else if (ConnectionState.SUSPENDED == connectionState) {//连接丢失,正在重试中,如果连接成功会话id不改变。如果会话id改变那么则是一个新的会话
            /**
             * (copy from org.apache.curator.framework.state.ConnectionState)
             * There has been a loss of connection. Leaders, locks, etc.
             * should suspend until the connection is re-established.
             * If the connection times-out you will receive a LOST notice
             *
             * ConnectionState.SUSPENDED=Watcher.Event.KeeperState.Disconnected即连接完全断开，curatorFramework不能自动重新连接,需要用户手动重新连接。
             */
            connected = false;
            //
            log.warn("ConnectionState.SUSPENDED:zk连接丢失");
            final long lastSessionId = getZooKeeperSessionId(curatorFramework);
            log.warn("初始会话id:" + this.sessionId + ",最新的会话id:" + lastSessionId);
            Runnable runnable = new Runnable() {
                //
                @Override
                public void run() {
                    //【设计思路】
                    // 假定zk连接断开的原因是因为网络瞬时异常，稍后还可以通过重连进行恢复。
                    // 如果是其他的异常情况 比如认证信息被修改，网络长时间断开。这种情况，不是框架重连可以修复。那么等待运维人员来进行解决。
                    // 当然如果真的出现了这种情况，执行器需要保证当前执行器的任务立即自动停止,避免出现问题。
                    // 本框架采用分布式续期机制,那么可以尽量避免分布式任务在执行过程中的数据重复处理问题。
                    do {
                        if (closed) {
                            break;
                        }
                        final long lastSessionId = getZooKeeperSessionId(curatorFramework);
                        log.warn("初始会话id:" + sessionId + ",最新的会话id:" + lastSessionId);
                        if (lastSessionId != sessionId) {
                            log.info("zk会话id改变,开始重启执行器");
                            restartExecutor();
                            break;
                        }
                        sleep();
                    } while (!closed && !connected);//closed或者connected退出循环。connected退出循环的触发点为:ConnectionState.RECONNECTED == connectionState
                }
            };
            //runnable对象创建成功后再进行检查closed,如果关闭了那么就不用重新重连了。
            if (!closed) {
                Thread thread = new Thread(runnable);
                thread.setDaemon(true);
                thread.start();
            }

        } else if (ConnectionState.LOST == connectionState) {
            /**
             *(copy from org.apache.curator.framework.state.ConnectionState)
             *The connection is confirmed to be lost. Close any locks, leaders, etc(需要注意的地方).
             * and attempt to re-create them.
             * NOTE: it is possible to get a RECONNECTED state after this
             * but you should still consider any locks, etc. as dirty/unstable
             *
             *ConnectionState.LOST=Watcher.Event.KeeperState.Expired
             */
            /**
             * 经过测试基本上到不了这个状态。当网络断开的时候ConnectionState.SUSPENDED == connectionState
             * session id变更检测线程检测到会话id=-1或者其他值，已经先进行执行器启动了。
             * 即使没有执行器启动这个过程。同样走不到这个状态。需要大家深刻理解这点。
             */
            log.warn("ConnectionState.LOST:zk连接过期,自动重新连接中,会创建新的会话id");
            //此时connected = false;
            final long sessionId = getZooKeeperSessionId(curatorFramework);
            log.warn("初始会话id:" + this.sessionId + ",最新的会话id:" + sessionId);
            log.info("会话过期,开始重启执行器");
            restartExecutor();
        } else if (ConnectionState.RECONNECTED == connectionState) {
            /**
             * (copy from org.apache.curator.framework.state.ConnectionState)
             * A suspended, lost, or read-only connection has been re-established
             *
             * ConnectionState.RECONNECTED=Watcher.Event.KeeperState.SyncConnected
             */
            connected = true;
            log.info("重新连接zk成功");
            final long lastSessionId = getZooKeeperSessionId(curatorFramework);
            log.warn("初始会话id:" + this.sessionId + ",最新的会话id:" + sessionId + "。两个会话id理论上不会被改变:" + (this.sessionId == lastSessionId) + ",要是不一致代表什么我也不知道???");
        } else if (ConnectionState.READ_ONLY == connectionState) {
            /**
             *(copy from org.apache.curator.framework.state.ConnectionState)
             *The connection has gone into read-only mode.
             * This can only happen if you pass true for CuratorFrameworkFactory.Builder.canBeReadOnly().
             * See the ZooKeeper doc regarding read only connections:
             * http://wiki.apache.org/hadoop/ZooKeeper/GSoCReadOnlyMode .
             * The connection will remain in read only mode until another state change is sent.
             *
             *ConnectionState.READ_ONLY=Watcher.Event.KeeperState.ConnectedReadOnly
             */
            log.warn("zk连接只读模式,该状态理论上不存在");
        } else {
            log.warn("其他状态:" + connectionState);
        }
    }

    private long getZooKeeperSessionId(CuratorFramework curatorFramework) {
        long sessionId;
        try {
            sessionId = curatorFramework.getZookeeperClient().getZooKeeper().getSessionId();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return -1;
        }
        return sessionId;
    }

    public void restartExecutor() {
        if (restartExecutored.compareAndSet(false, true)) {
            restartExecutor0();
        }
    }

    public abstract void restartExecutor0();

}